package com.dji.ux.sample.MQSignalClient;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import org.json.JSONException;
import org.json.JSONObject;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Description：rabbitMQ client
 * @Author： HaolinLv
 * @Date： 2022/1/9 17:24
 */

public class MQSignalClient {

    private String QUEUE_NAME;
    private MQEventListener mqEventListener;
    private Channel requestChannel;
    private DefaultConsumer mConsumer;


    public MQSignalClient(String queue_name, MQEventListener listener) {
        QUEUE_NAME = queue_name;
        mqEventListener = listener;
    }

    /**
     * 收消息（从发布者那边订阅消息）
     */
    public void startReceiveMQ() {
        new Thread(() -> {
            try {
                // 获取到连接
                //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
                requestChannel = RabbitMQUtils.getChannel();
                // 声明队列
                //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                /**
                 * 参数明细
                 * 1、queue 队列名称
                 * 2、durable 是否持久化，如果持久化，mq重启后队列还在
                 * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
                 * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
                 * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
                 */
                requestChannel.queueDeclare(QUEUE_NAME, true, false, false, null);
                //实现消费方法
                mConsumer = new DefaultConsumer(requestChannel) {
                    // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用

                    /**
                     * 当接收到消息后此方法将被调用
                     * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
                     * @param envelope 信封，通过envelope
                     * @param properties 消息属性
                     * @param body 消息内容
                     * @throws IOException
                     */
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        //交换机
                        //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                        long deliveryTag = envelope.getDeliveryTag();
                        // body 即消息体
                        String msg = new String(body, StandardCharsets.UTF_8);
                        System.out.println(" [x] received : " + msg + "!" + consumerTag);
                        JSONObject jsonMsg = null;
                        try {
                            jsonMsg = new JSONObject(msg);
                            String type = jsonMsg.getString("type");
                            switch (type) {
                                case "land_inspection_waypoint":
                                    mqEventListener.onMessage(deliveryTag, jsonMsg);
                                    break;
                                default:
                                    break;
                            }
                        } catch (JSONException e) {
                            e.printStackTrace();
                        }

                    }
                };

                //监听队列，第二个参数：是否自动进行消息确认。
                //参数：String queue, boolean autoAck, Consumer callback
                /**
                 * 参数明细：
                 * 1、queue 队列名称
                 * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
                 * 3、callback，消费方法，当消费者接收到消息要执行的方法
                 */
                requestChannel.basicConsume(QUEUE_NAME, true, mConsumer);
            } catch (Exception e) {
                e.printStackTrace();
                mqEventListener.onError(e.getMessage());
            }
        }).start();
    }

    /**
     * 发消息到MQ队列
     */
    public void sendMessage(String msg) {
        new Thread(() -> {
            try {
                // 获取到连接
                //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
                Channel channel = RabbitMQUtils.getChannel();
                channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                // 向指定的队列中发送消息
                //参数：String exchange, String routingKey, BasicProperties props, byte[] body
                /**
                 * 参数明细：
                 * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
                 * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
                 * 3、props，消息的属性
                 * 4、body，消息内容
                 */
                channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
                System.out.println(" [x] Sent '" + msg + "'");
            } catch (Exception e) {
                e.printStackTrace();
                mqEventListener.onError(e.getMessage());
            }
        }).start();
    }

    /**
     * 手动进行ACK
     * void basicAck(long deliveryTag, boolean multiple) throws IOException;
     * deliveryTag:用来标识消息的id
     * multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
     */
    public void ackMessage(long deliveryTag) {
        System.out.println(" [x] ackMessage '" + deliveryTag + "'");
        new Thread(() -> {
            try {
                Channel channel = mConsumer.getChannel();
                channel.basicAck(deliveryTag, false);
            } catch (Exception e) {
                e.printStackTrace();
                mqEventListener.onError(e.getMessage());
            }
        }).start();
    }

    public interface MQEventListener {

        void onMessage(long deliveryTag, JSONObject msg);

        void onError(String error);

    }

}
